package binlog.listener;

import binlog.model.TableId;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Hello world!
 *
 */
@Component
@Slf4j
public class DataEventListener implements BinaryLogClient.EventListener{
    /**
     * 参考-每一次的更新事件都有- [query,table_map,{insert,update,delete}] 三个连续的事件发生
     * io.debezium.connector.mysql.MySqlStreamingChangeEventSource#handleUpdateTableMetadata
     */
    private final Map<Long, TableId> tableIdsByTableNumber = new ConcurrentHashMap<>();

    @Override
    public void onEvent(Event event) {
        if (log.isInfoEnabled()){
            log.info("{}",event);
        }
        EventType eventType = event.getHeader().getEventType();
        EventData data = event.getData();
        switch (eventType){
            case ROTATE:{
                //TODO 新的binlog文件生成
                break;
            }
            case TABLE_MAP:{
                /**
                 * 存储表-元数据信息
                 */
                TableMapEventData metadata = (TableMapEventData) data;
                long tableId = metadata.getTableId();
                String databaseName = metadata.getDatabase();
                String tableName = metadata.getTable();
                TableId tableIdObj = new TableId(tableName,databaseName, tableName);
                tableIdsByTableNumber.put(tableId,tableIdObj);
                break;
            }
            case EXT_WRITE_ROWS:{
                WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
                long tableId = writeRowsEventData.getTableId();
                TableId tableMeta = tableIdsByTableNumber.get(tableId);
                List<Serializable[]> rows = writeRowsEventData.getRows();
                for (Serializable[] row : rows) {
                    Object pkObj = null;
                    pkObj = getPkObj(row[0], pkObj);
                    log.info("insert db: [{}] table: [{}], pk:[{}] , {}",tableMeta.getDatebaseName(),tableMeta.getTableName(),pkObj, Arrays.toString(row));
                }
                break;
            }
            case EXT_UPDATE_ROWS:{
                UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
                long tableId = updateRowsEventData.getTableId();
                TableId tableMeta = tableIdsByTableNumber.get(tableId);
                List<Map.Entry<Serializable[], Serializable[]>> rows = updateRowsEventData.getRows();
                for (Map.Entry<Serializable[], Serializable[]> row : rows) {
                    Serializable[] before = row.getKey();
                    Serializable[] after = row.getValue();

                    /**
                     * 一般来说第一个字段是主键
                     */
                    Serializable serializable = after[0];
                    Object pkObj = null;
                    pkObj = getPkObj(serializable, pkObj);

                    log.info("update db: [{}] table: [{}] pk: [{}] ",tableMeta.getDatebaseName(),tableMeta.getTableName(),pkObj);
                    log.info("before: {}", Arrays.toString(before));
                    log.info("after: {}",Arrays.toString(after));
                }
                break;
            }
            case EXT_DELETE_ROWS:{
                DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
                long tableId = deleteRowsEventData.getTableId();
                TableId tableMeta = tableIdsByTableNumber.get(tableId);
                List<Serializable[]> rows = deleteRowsEventData.getRows();
                for (Serializable[] row : rows) {
                    Object pkObj = null;
                    pkObj = getPkObj(row[0], pkObj);
                    log.info("delete db: [{}] table: [{}]  pk: [{}]  , {}",tableMeta.getDatebaseName(),tableMeta.getTableName(),pkObj, Arrays.toString(row));
                }
                break;
            }
            default:{
                //TODO 还是要记录偏移 postion
                log.info("ignore: {}",eventType.name());
            }
        }
    }

    private Object getPkObj(Serializable serializable, Object pkObj) {
        if (serializable instanceof Long){
            pkObj = (Long) serializable;
        }else if (serializable instanceof Integer){
            pkObj = (Integer) serializable;
        }else if (serializable instanceof String){
            pkObj = (String) serializable;
        }else {
            log.error("其他类型的主键:  [{}]", serializable);
        }
        return pkObj;
    }
}
